#ifdef _WIN32
#include <Windows.h>
#else
#include <unistd.h>
#endif

#include "ThreadPool.h"

using namespace Stray;
using namespace Stray::Core;

Request::Request(BindObject function)
	:function_(function)
{
}

void DefaultRequestQueue::put(const Request& request)
{
	queue_.push_back(request);
}

Request DefaultRequestQueue::take()
{
	Request output = queue_.front();
	queue_.pop_front();
	return output;
}

Bool DefaultRequestQueue::empty()
{
	if(queue_.empty()) return B_TRUE;
	else return B_FALSE;
}

WorkerThread::WorkerThread(ThreadPool *pThreadPool, const ThreadAttributes& threadAttri)
	:pThreadPool_(pThreadPool)
{
	BindObject func = bind(&WorkerThread::execute, this);
	Thread temp(threadAttri, func);
	thread_.swap(temp);
}

void WorkerThread::execute()
{
	for(;;)
	{
		if(pThreadPool_->isTerminating())
		{
			break;
		}

		Request request = pThreadPool_->takeRequest(condition_);
		request.Execute();
	}

	//std::cout << "WorkerThread finished." << std::endl;
}

void WorkerThread::wait()
{
	thread_.wait();
}

ThreadPool::ThreadPool(size_t threads, const AttributesArray& attributes)
	:isTerminating_(B_FALSE)
	,hasTerminated_(B_FALSE)
{
	assert(threads == attributes.size());
	for(size_t i = 0; i < threads; ++i)
	{
		threads_.push_back(new WorkerThread(this, attributes[i]));
	}
}

ThreadPool::~ThreadPool()
{
	terminate();

	Stray::Vector<WorkerThread*>::type::iterator iter;
	for(iter = threads_.begin(); iter != threads_.end(); ++iter)
	{
		delete (*iter);
	}
}

void ThreadPool::putRequest(const Request& request)
{
	ScopedLock lock(mutex_);
	requestQueue_.put(request);

	Stray::Vector<Condition*>::type::iterator iter;
	for(iter = conditions_.begin(); iter != conditions_.end(); ++iter)
	{
		(*iter)->notify();
	}
	conditions_.clear();
}

Request ThreadPool::takeRequest(Condition& condition)
{
	if(isTerminating_) return Request(BindObject());
	ScopedLock lock(mutex_);
	if(requestQueue_.empty())
	{
		conditions_.push_back(&condition);
		condition.wait(lock);
	}

	if(requestQueue_.empty()) return Request(BindObject());
	return requestQueue_.take();
}

void ThreadPool::terminate()
{
	if(hasTerminated_ == B_TRUE) return;

#ifdef _WIN32
	Sleep(100);
#else
	sleep(0.1);
#endif

	{
		ScopedLock lock(mutex_);

		isTerminating_ = B_TRUE;
	
		Stray::Vector<Condition*>::type::iterator i;
		for(i = conditions_.begin(); i != conditions_.end(); ++i)
		{
			(*i)->notify();
		}
	}
	
	Stray::Vector<WorkerThread*>::type::iterator iter;
	for(iter = threads_.begin(); iter != threads_.end(); ++iter)
	{
		(*iter)->wait();
	}
	
	hasTerminated_ = B_TRUE;
}

Bool ThreadPool::isTerminating()
{
	return isTerminating_;
}
